﻿using CommonModels;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using System.Net.Http;
using System.Net.Http.Formatting;
using System.Diagnostics;

namespace OrderProcessing
{
    class Program
    {
        private static int _counter;
        private static int _loops;
        static void Main(string[] args)
        {
            BroadcastBlockRun();
        }

        static void BroadcastBlockRun()
        {
            Stopwatch sw = new Stopwatch();
            sw.Start();
            BroadcastBlock<int> bclk = new BroadcastBlock<int>(i => i, new DataflowBlockOptions { BoundedCapacity = 10 });
            int ablkCount = 2;
            int tblkCount = 2;
            ActionBlock<int>[] ablkArray = new ActionBlock<int>[ablkCount];
            TransformBlock<int, int>[] tblkArray = new TransformBlock<int, int>[tblkCount];
            for (int i = 0; i < ablkCount; i++)
            {
                int index = i;
                ///async and proper MaxDegreeOfParallelism gets performance improvement
                ///
                ActionBlock<int> ablk = new ActionBlock<int>(
                    async (j) =>
                    {
                        await Task.Delay(1000);
                        Console.WriteLine("ablk" + index + " got " + j + " at " + DateTime.Now.ToString("HH:mm:ss.fff") + " in thread " + Thread.CurrentThread.ManagedThreadId);
                    }, new ExecutionDataflowBlockOptions
                    {
                        BoundedCapacity = 5,
                        MaxDegreeOfParallelism = 1
                    });
                ablkArray[i] = ablk;
                bclk.LinkTo(ablk, (m)=>m%2==0);                
                bclk.Completion.ContinueWith((t) =>
                {
                    ablk.Complete();
                });
            }
            for (int i = 0; i < 10; i++)
            {
                SpinWait.SpinUntil(() =>
                {
                    var result = bclk.Post(i);
                    if (!result)
                    {
                        Console.WriteLine("wait for buffer");
                    }
                    return result;
                });
            }
            //Thread.Sleep(2);
            //ActionBlock<int> ablk_new = new ActionBlock<int>(
            //        async (j) =>
            //        {
            //            await Task.Delay(1000);
            //            Console.WriteLine("Broadcast is completed:" +
            //            bclk.Completion.IsCompleted + ". ablk_new got " + j + " at " + DateTime.Now.ToString("HH:mm:ss.fff") + " in thread " + Thread.CurrentThread.ManagedThreadId);
            //        }, new ExecutionDataflowBlockOptions
            //        {
            //            BoundedCapacity = 10,
            //            MaxDegreeOfParallelism = -1
            //        });
            //bclk.LinkTo(ablk_new);
            //bclk.Completion.ContinueWith((t) =>
            //{
            //    ablk_new.Complete();
            //});
            bclk.Complete();
            Console.WriteLine("Broadcast wait for completion at " + sw.ElapsedMilliseconds);
            bclk.Completion.Wait();
            Console.WriteLine(string.Format("Broadcast block done in {0} ms!", sw.ElapsedMilliseconds));

            Task.WaitAll(ablkArray.Select(a => a.Completion).ToArray());
            Console.WriteLine(string.Format("ablk block done in {0} ms!", sw.ElapsedMilliseconds));
            sw.Stop();

            Console.ReadLine();
        }

        private static void BlockRun()
        {
            Stopwatch sw = new Stopwatch();
            sw.Start();
            //capacity for throttling
            BufferBlock<int> bblk = new BufferBlock<int>(new DataflowBlockOptions { BoundedCapacity = 100000 });
            
            int ablkCount = 2;
            ActionBlock<int>[] ablkArray = new ActionBlock<int>[ablkCount];
            for (int i = 0; i < ablkCount; i++)
            {
                int index = i;
                ///async and proper MaxDegreeOfParallelism gets performance improvement
                ///
                ActionBlock<int> ablk = new ActionBlock<int>(
                    async (j) =>
                    {
                        await Task.Delay(1000);
                        Console.WriteLine("ablk" + index + " got " + j + " at " + DateTime.Now + " in thread " + Thread.CurrentThread.ManagedThreadId);
                    }, new ExecutionDataflowBlockOptions
                    {
                        BoundedCapacity = 20000,
                        MaxDegreeOfParallelism = -1
                    });
                ablkArray[i] = ablk;
                bblk.LinkTo(ablk,m=>m%2==0);
                bblk.Completion.ContinueWith((t) =>
                {
                    ablk.Complete();
                });
            }
            for (int i = 0; i < 100000; i++)
            {
                SpinWait.SpinUntil(() =>
                {
                    var result = bblk.Post(i);
                    if (!result)
                    {
                        Console.WriteLine("wait for buffer");
                    }
                    return result;
                });
            }
            bblk.Complete();
            Console.WriteLine("wait for completion");
            bblk.Completion.Wait();
            Console.WriteLine(string.Format("buffer block done in {0} ms!", sw.ElapsedMilliseconds));
            //ablk1.Completion.Wait();
            //ablk2.Completion.Wait();
            //ablk3.Completion.Wait();
            Task.WaitAll(ablkArray.Select(a => a.Completion).ToArray());
            Console.WriteLine(string.Format("ablk block done in {0} ms!", sw.ElapsedMilliseconds));
            sw.Stop();
            Console.ReadLine();
        }

        private static void ProcessOrderMultiBlock()
        {
            try
            {
                ServicePointManager.DefaultConnectionLimit = int.MaxValue;
                Console.Write("Wait for services to start, then enter the desired degree of parallelism: ");
                var degreeOfParallellism = int.Parse(Console.ReadLine());
                Console.Write("Wait for services to start, then enter the desired loops: ");
                _loops = int.Parse(Console.ReadLine());
                Console.Write("Wait for services to start, then enter the desired throttling: ");
                var throttling = int.Parse(Console.ReadLine());
                Console.WriteLine();

                var asyncPersistAction = new Func<ItemTrackingDetail, Task>(PersistToDatabase);

                //var tupleAsyncPersistAction = new Action<Tuple<ItemTrackingDetail,ItemTrackingDetail>>((tuple)=>
                //    {
                //        Parallel.Invoke(()=> asyncPersistAction(tuple.Item1),()=>asyncPersistAction(tuple.Item2));
                //    }
                //);


                //define the blocks
                var orderBuffer = new BufferBlock<Order>(new ExecutionDataflowBlockOptions { BoundedCapacity = throttling });//new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = degreeOfParallellism });
                var broadcaster = new BroadcastBlock<Order>(order => order);//, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = degreeOfParallellism });
                var upsBatcher = new BatchBlock<Order>(5);
                var fedexBatcher = new BatchBlock<Order>(5);


                var storageProcessor = new ActionBlock<ItemTrackingDetail>(asyncPersistAction, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = degreeOfParallellism });
                //var tupleStorageProcessor = new ActionBlock<Tuple<ItemTrackingDetail, ItemTrackingDetail>>(tupleAsyncPersistAction, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = degreeOfParallellism * 10 });



                //link the blocks together
                orderBuffer.LinkTo(broadcaster);
                broadcaster.LinkTo(upsBatcher, order => order.Carrier == CarrierType.Ups);
                broadcaster.LinkTo(fedexBatcher, order => order.Carrier == CarrierType.Fedex);
                List<TransformManyBlock<Order[], ItemTrackingDetail>> upsProcessorList = new List<TransformManyBlock<Order[], ItemTrackingDetail>>();
                List<TransformManyBlock<Order[], ItemTrackingDetail>> fedexProcessorList = new List<TransformManyBlock<Order[], ItemTrackingDetail>>();
                for (int i = 0; i < degreeOfParallellism; i++)
                {
                    var upsProcessor = new TransformManyBlock<Order[], ItemTrackingDetail>(orders => { return PostToCarrierAsync(CarrierType.Ups, orders); }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = degreeOfParallellism });
                    upsProcessorList.Add(upsProcessor);
                    upsBatcher.LinkTo(upsProcessor);
                    upsProcessor.LinkTo(storageProcessor);
                    var fedexProcessor = new TransformManyBlock<Order[], ItemTrackingDetail>(orders => { return PostToCarrierAsync(CarrierType.Fedex, orders); }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = degreeOfParallellism });
                    fedexProcessorList.Add(fedexProcessor);
                    fedexBatcher.LinkTo(fedexProcessor);
                    fedexProcessor.LinkTo(storageProcessor);

                }
                Task[] processors = upsProcessorList.Select(p => p.Completion).Union(fedexProcessorList.Select(p => p.Completion)).ToArray();
                Action<Task> postOrderCompletion = t =>
                {
                    Task.WaitAll(processors);
                    storageProcessor.Complete();
                };
                upsProcessorList.ForEach(p => p.Completion.ContinueWith(postOrderCompletion));
                fedexProcessorList.ForEach(p => p.Completion.ContinueWith(postOrderCompletion));


                //set the completion propagation rules
                orderBuffer.Completion.ContinueWith(t =>
                {
                    broadcaster.Complete();
                });

                broadcaster.Completion.ContinueWith(t =>
                {
                    upsBatcher.Complete();
                    fedexBatcher.Complete();
                });

                upsBatcher.Completion.ContinueWith(t => upsProcessorList.AsParallel().ForAll(p => p.Complete()));
                fedexBatcher.Completion.ContinueWith(t => fedexProcessorList.AsParallel().ForAll(p => p.Complete()));

                //upsProcessor.Completion.ContinueWith(t => joinProcessor.Complete());
                //fedexProcessor.Completion.ContinueWith(t => joinProcessor.Complete());

                //joinProcessor.Completion.ContinueWith(t =>
                //{
                //    Task.WaitAll(upsProcessor.Completion, fedexProcessor.Completion);
                //    tupleStorageProcessor.Complete();
                //});

                var s = Stopwatch.StartNew();
                for (int i = 0; i < _loops; i++)
                {
                    bool posted = false;

                    while (!posted)
                    {
                        posted = orderBuffer.Post(
                                new Order
                                {
                                    Carrier = i % 2 == 0 ? CarrierType.Ups : CarrierType.Fedex,
                                    OrderId = Guid.NewGuid(),
                                    Items = new List<OrderItem>()
                                });
                        if (!posted)
                        {
                            Console.WriteLine("buffer is full, wait 100ms");
                            Thread.Sleep(100);
                        }
                    }
                }

                orderBuffer.Complete();
                storageProcessor.Completion.Wait();
                //tupleStorageProcessor.Completion.Wait();

                s.Stop();

                Console.WriteLine("cargos shipped:{0}", _counter);
                Console.WriteLine("Processing completed in {0}. Press <Enter> to exit.", s.Elapsed);
                Console.ReadLine();
            }
            catch (Exception ex)
            {
                Console.WriteLine("Error occured in main due to:" + ex);
            }
        }

        private static void ProcessOrder()
        {
            try
            {
                ServicePointManager.DefaultConnectionLimit = int.MaxValue;
                Console.Write("Wait for services to start, then enter the desired degree of parallelism: ");
                var degreeOfParallellism = int.Parse(Console.ReadLine());
                Console.Write("Wait for services to start, then enter the desired loops: ");
                _loops = int.Parse(Console.ReadLine());
                Console.Write("Wait for services to start, then enter the desired throttling: ");
                var throttling = int.Parse(Console.ReadLine());
                Console.WriteLine();

                var asyncPersistAction = new Func<ItemTrackingDetail, Task>(PersistToDatabase);

                //var tupleAsyncPersistAction = new Action<Tuple<ItemTrackingDetail,ItemTrackingDetail>>((tuple)=>
                //    {
                //        Parallel.Invoke(()=> asyncPersistAction(tuple.Item1),()=>asyncPersistAction(tuple.Item2));
                //    }
                //);


                //define the blocks
                var orderBuffer = new BufferBlock<Order>(new ExecutionDataflowBlockOptions { BoundedCapacity = throttling });//new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = degreeOfParallellism });
                var broadcaster = new BroadcastBlock<Order>(order => order);//, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = degreeOfParallellism });
                var upsBatcher = new BatchBlock<Order>(5);
                var fedexBatcher = new BatchBlock<Order>(5);
                var upsProcessor = new TransformManyBlock<Order[], ItemTrackingDetail>(orders => { return PostToCarrierAsync(CarrierType.Ups, orders); }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = degreeOfParallellism });
                var fedexProcessor = new TransformManyBlock<Order[], ItemTrackingDetail>(orders => { return PostToCarrierAsync(CarrierType.Fedex, orders); }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = degreeOfParallellism });

                //var joinProcessor = new JoinBlock<ItemTrackingDetail, ItemTrackingDetail>(new GroupingDataflowBlockOptions() { Greedy = false });

                var storageProcessor = new ActionBlock<ItemTrackingDetail>(asyncPersistAction, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = degreeOfParallellism });
                //var tupleStorageProcessor = new ActionBlock<Tuple<ItemTrackingDetail, ItemTrackingDetail>>(tupleAsyncPersistAction, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = degreeOfParallellism * 10 });



                //link the blocks together
                orderBuffer.LinkTo(broadcaster);
                broadcaster.LinkTo(upsBatcher, order => order.Carrier == CarrierType.Ups);
                broadcaster.LinkTo(fedexBatcher, order => order.Carrier == CarrierType.Fedex);
                upsBatcher.LinkTo(upsProcessor);
                fedexBatcher.LinkTo(fedexProcessor);
                upsProcessor.LinkTo(storageProcessor);
                fedexProcessor.LinkTo(storageProcessor);

                //upsProcessor.LinkTo(joinProcessor.Target1);
                //fedexProcessor.LinkTo(joinProcessor.Target2);
                //joinProcessor.LinkTo(tupleStorageProcessor);

                //set the completion propagation rules
                orderBuffer.Completion.ContinueWith(t =>
                {
                    broadcaster.Complete();
                });

                broadcaster.Completion.ContinueWith(t =>
                {
                    upsBatcher.Complete();
                    fedexBatcher.Complete();
                });

                upsBatcher.Completion.ContinueWith(t => upsProcessor.Complete());
                fedexBatcher.Completion.ContinueWith(t => fedexProcessor.Complete());

                Action<Task> postOrderCompletion = t =>
                {
                    Task.WaitAll(upsProcessor.Completion, fedexProcessor.Completion);
                    storageProcessor.Complete();
                };
                upsProcessor.Completion.ContinueWith(postOrderCompletion);
                fedexProcessor.Completion.ContinueWith(postOrderCompletion);

                //upsProcessor.Completion.ContinueWith(t => joinProcessor.Complete());
                //fedexProcessor.Completion.ContinueWith(t => joinProcessor.Complete());

                //joinProcessor.Completion.ContinueWith(t =>
                //{
                //    Task.WaitAll(upsProcessor.Completion, fedexProcessor.Completion);
                //    tupleStorageProcessor.Complete();
                //});

                var s = Stopwatch.StartNew();
                for (int i = 0; i < _loops; i++)
                {
                    bool posted = false;

                    while (!posted)
                    {
                        posted = orderBuffer.Post(
                                new Order
                                {
                                    Carrier = i % 2 == 0 ? CarrierType.Ups : CarrierType.Fedex,
                                    OrderId = Guid.NewGuid(),
                                    Items = new List<OrderItem>()
                                });
                        if (!posted)
                        {
                            Console.WriteLine("buffer is full, wait 100ms");
                            Thread.Sleep(100);
                        }
                    }
                }

                orderBuffer.Complete();
                storageProcessor.Completion.Wait();
                //tupleStorageProcessor.Completion.Wait();

                s.Stop();

                Console.WriteLine("cargos shipped:{0}", _counter);
                Console.WriteLine("Processing completed in {0}. Press <Enter> to exit.", s.Elapsed);
                Console.ReadLine();
            }
            catch (Exception ex)
            {
                Console.WriteLine("Error occured in main due to:" + ex);
            }
        }


        // Simulates transformation of a list of Order objects into a service api model (ShipDetail).
        private static List<ShipDetail> CreateShipDetails(IEnumerable<Order> orders)
        {
            try
            {
                var shipDetails = orders.Select(order =>
                        new ShipDetail
                        {
                            ShipId = order.OrderId,
                            Items = order.Items.Select(item =>
                                new ShipItemInfo
                                {
                                    Sku = item.Sku,
                                }).ToList()
                        });

                return shipDetails.ToList();
            }
            catch (Exception ex)
            {
                Console.WriteLine("CreateShipDetails failed due to:" + ex);
            }
            return default(List<ShipDetail>);
        }

        // Sends orders to a shipping service endpoint dependent on the specified carrier
        private static async Task<IEnumerable<ItemTrackingDetail>> PostToCarrierAsync(CarrierType carrierType, Order[] orders)
        {
            try
            {
                var shipDetails = CreateShipDetails(orders);
                //Console.WriteLine("Sending {0} orders to {1}.", orders.Length, carrierType);
                string url = "http://arch-lpt-win-02.qa.nt.ctripcorp.com:9099/api/";//"http://localhost:3001/api/";//"http://arch-lpt-win-02.qa.nt.ctripcorp.com:3001/api/";// 
                var client = new System.Net.Http.HttpClient();
                var response = await client.PostAsync(url + carrierType.ToString(), shipDetails, new JsonMediaTypeFormatter());
                if (!response.IsSuccessStatusCode)
                    throw new ApplicationException("Exception in shipping processor (" + carrierType + "): " + response.ReasonPhrase);

                var results = await response.Content.ReadAsAsync<List<ItemTrackingDetail>>();
                return results;
            }
            catch (Exception ex)
            {
                Console.WriteLine("PostToCarrierAsync failed due to:" + ex);
            }
            return default(IEnumerable<ItemTrackingDetail>);
        }

        // Simulates persistence of tracking numbers for each item to a database.
        private static async Task PersistToDatabase(ItemTrackingDetail itemTrackingDetail)
        {
            // ...  your DB code here
            //Simulate updating the order to the database.
            await Task.Delay(50);
            //Console.WriteLine("Wrote tracking details to DB for order{0}.", itemTrackingDetail.ShippingId);
            Interlocked.Add(ref _counter, 1);
            if (_counter % 10000 == 0)
            {
                Console.WriteLine("cargos shipped:{0}", _counter);
            }
        }

        private static async Task TuplePersistToDatabase(ItemTrackingDetail itemTrackingDetail)
        {
            // ...  your DB code here
            //Simulate updating the order to the database.
            await Task.Delay(50);
            Console.WriteLine("Wrote tracking details to DB for order{0}.", itemTrackingDetail.ShippingId);
            //Interlocked.Add(ref _counter, 1);
        }
    }
}
